package com.yc.project2_knn.group2;

import com.yc.project2_knn.EuclideanDistanceCalculator;
import com.yc.project2_knn.bean.BankMarketing;
import com.yc.project2_knn.bean.Distance;
import com.yc.project2_knn.bean.Sample;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * knn算法的任务调度类;
 */
public class ParallerroupKnnClassfier {
    private int k; //knn的K ,标识近邻几个
    private int numThread;//线程数
    private boolean paralleSort; //排序石佛营并行
    private List<BankMarketing> dataSet;

    private ThreadPoolExecutor executor;//引入线程池

    public ParallerroupKnnClassfier(int k, int numThread, boolean paralleSort, List<BankMarketing> dataSet) {
        this.k = k;
        this.numThread = numThread;
        this.paralleSort = paralleSort;
        this.dataSet = dataSet;
        this.executor = (ThreadPoolExecutor) Executors.newFixedThreadPool((int) (this.numThread * 1.5));
    }


    public String classify(Sample example) throws InterruptedException {
        CountDownLatch endController = new CountDownLatch(numThread);//计数的类  计数器 -》0
        //endController.countDown();// 让线程--
        Distance[] distances = new Distance[dataSet.size()];
        //1.计算12个线程中每 个线程它的计算任务的startIndex, endIndex
        int length = dataSet.size() / numThread; //分段区间：
        int startIndex = 0;
        int endIndex = length;
        //List<Thread> list = new ArrayList<>();
        //2.根据numThreads 创建任务， 并绑定到线程上
        for (int i = 0; i < numThread; i++) {
            ////计算example这个条测试数据与 dataSet中39129条数据的距离(只要计算startIndex-endIndex )，将距离的结果存到distances
            GroupDiatanceTask task = new GroupDiatanceTask(distances, startIndex, endIndex, dataSet,
                    example,new EuclideanDistanceCalculator(),endController);
//            Thread t =new Thread(task);
//            t.start();
//            list.add(t);
            executor.execute(task); ////加入线程池后，如何让主程序停下来，

           // System.out.println("第" + i + "个线程的计算范围为：" + startIndex + "-" + endIndex);
            //计算下一个线程的范围
            startIndex = endIndex;
            //rows1=2000
            endIndex = i == numThread - 2 ? dataSet.size() : endIndex + length;

        }
        //3.调用每个线程 join()， 让主线程停止，好计算时间
//        for(Thread t:list){
//            t.join();
//        }
        endController.await();//相当于上面的join   阻塞，当这个 countdownZatch中的值为日， 则解除阻塞，

        //4.排序距离
        if (paralleSort) {  //空指针异常 线程在线程池里的 他还没有跑完算距离，主程序已经在排序了
            Arrays.parallelSort(distances);
        } else {
            Arrays.sort(distances);
        }
        //5.将前k个样本的标签 存到一个Map<String, Integer>
        Map<String, Integer> results = new HashMap<>();
        for (int i = 0; i < k; i++) {
            Sample sample = dataSet.get(distances[i].getIndex());
            // sample getTag() -> 获取标签getData()
            String tag = sample.getTag();
//            if( results.containsKey( tag) ){
//                results.put( tag, results.get(tag)+1 );
//            }else{
//                results.put(tag, 1);
//            }
            results.merge(tag, 1, (a, b) -> a + b);
        }

        //6.取出map中次数最多的标签名返回
        //传统写法
//        Set<Map.Entry<String ,Integer>> set=results.entrySet();
//        int max=0;
//        String maxTag="";
//        for( Map. Entry<String, Integer> entry: set ){
//            String tag= entry.getKey();
//            int value=entry.getValue();
//            if(value>max){
//                maxTag= tag;
//            }
//        }
//        return maxTag;
        //// Collections 工具类对集合进行操作(排序，查找，最值....
        return Collections.max(results.entrySet(), Map.Entry.comparingByValue()).getKey();//根据值来比;
    }

    public void destroy() {
        if (executor != null) {
            executor.shutdown();
        }
    }
}
